package cn.me.alphamq.nameserver.handler.service;

import cn.me.alphamq.common.entity.req.ListAllTopicInfoReq;
import cn.me.alphamq.common.entity.req.PullBrokerAddressReq;
import cn.me.alphamq.common.entity.req.ServiceHeartBeatReq;
import cn.me.alphamq.common.entity.req.ServiceRegistryReq;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.EventBus;
import cn.me.alphamq.common.event.model.Event;
import cn.me.alphamq.common.util.NettyUtil;
import cn.me.alphamq.nameserver.event.model.service.*;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * TcpMsg 对象的处理器，可以被所有线程共享
 *
 * @author f
 */
@ChannelHandler.Sharable
public class TcpMsgServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger log = LoggerFactory.getLogger(TcpMsgServerHandler.class);

    /**
     * 事件总线
     */
    private final EventBus eventBus;

    public TcpMsgServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        // 根据 code 的不同，将 body 解析成不同的 Event 对象
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();

        Event event;
        if (EventCodeEnum.REGISTRY_REQ.getCode() == code) {
            ServiceRegistryReq serviceRegistryReq = JSON.parseObject(body, ServiceRegistryReq.class);
            // 如果注册信息没有传递 ip 和 port，使用连接的 ip 和 port 来赋值
            if (serviceRegistryReq.getIp() == null) {
                InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
                serviceRegistryReq.setIp(inetSocketAddress.getHostString());
                serviceRegistryReq.setPort(inetSocketAddress.getPort());
            }
            event = new RegistryEvent(serviceRegistryReq);
        } else if (EventCodeEnum.HEART_BEAT_REQ.getCode() == code) {
            event = new HeartBeatEvent(JSON.parseObject(body, ServiceHeartBeatReq.class));
        } else if (EventCodeEnum.PULL_BROKER_ADDRESS_REQ.getCode() == code) {
            event = new PullBrokerAddressEvent(JSON.parseObject(body, PullBrokerAddressReq.class));
        } else if (EventCodeEnum.LIST_ALL_TOPIC_INFO_REQ.getCode() == code) {
            event = new ListAllTopicInfoEvent(JSON.parseObject(body, ListAllTopicInfoReq.class));
        } else {
            throw new RuntimeException("unknown event code:" + code);
        }

        // 给事件绑定 ctx 对象
        event.setCtx(ctx);

        // 发布事件
        eventBus.publish(event);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        // 如果依赖任务剔除节点，会有三个心跳周期的延迟，本方法做到了链接断开立马剔除的效果
        log.info("service instance [{}] inactive...", NettyUtil.getCtxId(ctx));
        UnRegistryEvent unRegistryEvent = new UnRegistryEvent();
        unRegistryEvent.setCtx(ctx);
        eventBus.publish(unRegistryEvent);
    }
}
